package de.preissler.dataimport;

import java.io.File;
import java.util.LinkedList;
import java.util.List;

import jxl.Cell;
import jxl.CellType;
import jxl.NumberCell;
import jxl.Sheet;
import jxl.Workbook;
import jxl.WorkbookSettings;

import me.prettyprint.cassandra.serializers.DoubleSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;

/**
 * 
 * @author Eric Preissler
 *
 */
public class XLSImport
{
	/*
	 * Der Basis Wert fuer die Index Formel
	 */
	private static final int BASE = 1000;
	
	/*
	 * Name des Sheets wo die Daten sind
	 */
	private static final String DATA = "Data";
	
	/*
	 * Pre- und Postfix welche das Datum im Dateinamen umschliessen
	 */
	private static final String PRE = "ICR.";
	private static final String POST = ".xls";
	
	/*
	 * Zelleninhalt mit dem die Kopfzeile bestimmt wird
	 */
	private static final String HEAD = "Index Trading Symbol";
	
	/*
	 * Namen der Spalten die fuer die Berechnung relevant sind
	 */
	private static final String INDEX_ISIN = "Index ISIN";
	private static final String ISIN = "ISIN";
	private static final String KT = "Kt";
	private static final String PI0 = "pi0";
	private static final String QI0 = "qi0";
	private static final String QIT = "qit";
	private static final String FFI = "ffit";
	private static final String CI = "ci";
	
	private Cluster cluster;
	private Keyspace keyspace;
	private String xlsPath;
	
	/**
	 * Erstellt ein XLS Import Objekt
	 * @param cluster Cassandra-Cluster
	 * @param keyspace Exchange Keyspace
	 * @param xlsPath Pfad zu den Dateien
	 */
	public XLSImport(Cluster cluster, Keyspace keyspace, String xlsPath)
	{
		this.cluster = cluster;
		this.keyspace = keyspace;
		this.xlsPath = xlsPath;
	}

	/**
	 * Kopiert alle xls Dateien in die Datenbank
	 * @throws Exception
	 */
	public void copyAllToCassandra() throws Exception
	{
		//Pfade der auszulesenden Dateien besorgen
		List<String> filePathList = getFilePathList(xlsPath);
		
		//Jede Datei einzeln auslesen
		for(String path : filePathList)
		{
			File file = new File(path);
			MyUtil.trace("Working on \"" + file.getName() + "\" (" + (filePathList.indexOf(path)+1) + "/" + filePathList.size() + ")!");
			copyFileToCassandra(cluster, keyspace, file);
		}
		
	}
	
	/*
	 * Gibt eine Liste mit den Dateipfaden zurueck
	 */
	private List<String> getFilePathList(String xlsPath)
	{
		File[] fileArray = new File(xlsPath).listFiles();
		List<String> fileList = new LinkedList<String>();
		
		for(File f : fileArray)
		{
			if(f.isFile() && f.getName().endsWith(".xls"))
			{
				fileList.add(f.getPath());
			}
		}
		
		MyUtil.trace(fileList.size() + " .xls files has been found!");
		
		return fileList;
	}
	
	/*
	 * Kopiert eine .csv Datei zu Cassandra
	 */
	private void copyFileToCassandra(Cluster cluster, Keyspace keyspace, File file) throws Exception
	{	
		//Mutator holen, mit ihm werden Daten zu Cassandra geschrieben
		Mutator<String> mutator = HFactory.createMutator(keyspace, MyUtil.SS);
		
		//Laedt die Datei in den RAM und macht sie zugreifbar ueber die JExcel API
		WorkbookSettings settings = new WorkbookSettings();
		settings.setSuppressWarnings(true);
		Workbook workbook = Workbook.getWorkbook(file, settings);
		
		//Laedt Informationen fuer das navigieren der xls
		XLSMap xlsMap = getXLSMap(workbook, file.getName());
		
		//Index Zeile erzeugen
		saveIndexRecord(xlsMap, mutator);
		
		//Jede Zeile ist eine SuperColumn
		for(int row = xlsMap.startLine; row < xlsMap.endLine; row++)
		{
			List<HColumn<String, Double>> list = new LinkedList<HColumn<String, Double>>();
			
			//ISIN
			String isin = xlsMap.sheet.getCell(xlsMap.isinColumn, row).getContents();
			
			//schlusskurs in 0
			double p0;
			Cell cell = xlsMap.sheet.getCell(xlsMap.pi0Column, row);
			if(cell.getType() == CellType.NUMBER)
			{
				p0 = ((NumberCell)cell).getValue();
			}
			else
			{
				p0 = -1;
			}
			HColumn<String, Double> c = HFactory.createColumn("price at zero", p0,
					StringSerializer.get(), DoubleSerializer.get());
			list.add(c);
			
			//aktien in 0
			double q0;
			cell = xlsMap.sheet.getCell(xlsMap.qi0Column, row);
			if(cell.getType() == CellType.NUMBER)
			{
				q0 = ((NumberCell)cell).getValue();
			}
			else
			{
				q0 = -1;
			}
			c = HFactory.createColumn("#shares at zero", q0,
					StringSerializer.get(), DoubleSerializer.get());
			list.add(c);
			
			//aktien in t
			double qt;
			cell = xlsMap.sheet.getCell(xlsMap.qitColumn, row);
			if(cell.getType() == CellType.NUMBER)
			{
				qt = ((NumberCell)cell).getValue();
			}
			else
			{
				qt = -1;
			}
			c = HFactory.createColumn("#shares at t", qt,
					StringSerializer.get(), DoubleSerializer.get());
			list.add(c);
			
			//fff
			double fff;
			cell = xlsMap.sheet.getCell(xlsMap.ffiColumn, row);
			if(cell.getType() == CellType.NUMBER)
			{
				fff = ((NumberCell)cell).getValue();
			}
			else
			{
				fff = -1;
			}
			c = HFactory.createColumn("free float", fff, StringSerializer.get(), DoubleSerializer.get());
			list.add(c);
			
			//correction
			double co;
			cell = xlsMap.sheet.getCell(xlsMap.ciColumn, row);
			if(cell.getType() == CellType.NUMBER)
			{
				co = ((NumberCell)cell).getValue();
			}
			else
			{
				co = -1;
			}
			c = HFactory.createColumn("correction", co, StringSerializer.get(), DoubleSerializer.get());
			list.add(c);
			
			
			//Nun wird die Column unter der Zeit Zeile gespeichert
			HSuperColumn<String, String, Double> sc = HFactory.createSuperColumn(isin, list, StringSerializer.get(),
					StringSerializer.get(), DoubleSerializer.get());
			mutator.insert(xlsMap.date, xlsMap.symbol, sc);
		}
		
		workbook.close();
	}
	
	/*
	 * Holt Informationen ueber die XLS Datei
	 */
	private XLSMap getXLSMap(Workbook workbook, String fileName) throws Exception
	{
		//Symbol herausfiltern
		String symbol = fileName.substring(0, fileName.indexOf("_ICR"));
		if(symbol.equals("TecDAX"))
			symbol = "TDXP";
		
		//Datum herausfiltern
		String date = fileName.substring(fileName.indexOf(PRE)+PRE.length(), fileName.lastIndexOf(POST));
		date = date.substring(0, 4) + "-" + date.substring(4, 6) + "-" + date.substring(6);
		
		Sheet sheet = workbook.getSheet(DATA);
		
		//Zeilenkopf finden
		int headRow = -1;
		for(int row = 0; row < sheet.getRows(); row++)
		{
			if(sheet.getCell(0, row).getContents().equals(HEAD))
			{
				headRow = row;
				break;
			}
		}
		if(headRow == -1)
			throw new Exception("Found no head row!");
		
		//Spalten finden die den erfoderlichen Inhalt haben
		int kt = -1;
		int indexIsin = -1;
		int isin = -1;
		int pi0 = -1;
		int qit = -1;
		int qi0 = -1;
		int ffi = -1;
		int ci = -1;
		for(int column = 0; column < sheet.getColumns(); column++)
		{
			String cellContent = sheet.getCell(column, headRow).getContents();
			
			if(kt == -1 && cellContent.startsWith(KT))
				kt = column;
			
			if(indexIsin == -1 && cellContent.equals(INDEX_ISIN))
				indexIsin = column;
			
			if(isin == -1 && cellContent.equals(ISIN))
				isin = column;
			
			if(pi0 == -1 && cellContent.startsWith(PI0))
				pi0 = column;
			
			if(qit == -1 && cellContent.startsWith(QIT))
				qit = column;
			
			if(qi0 == -1 && cellContent.startsWith(QI0))
				qi0 = column;
			
			if(ffi == -1 && cellContent.startsWith(FFI))
				ffi = column;
			
			if(ci == -1 && cellContent.startsWith(CI))
				ci = column;
			
			if(kt != -1 && indexIsin != -1 && isin != -1 && ci != -1 && ffi != -1 && qi0 != -1 && qit != -1 && pi0 != -1)
				break;
		}
		if(!(kt != -1 && indexIsin != -1 && isin != -1 && ci != -1 && ffi != -1 && qi0 != -1 && qit != -1 && pi0 != -1))
			throw new Exception("Important column could not be found!");
		
		//Zeile finden wo der Inhalt los geht
		int startOfContent = -1;
		for(int row = 0; row < sheet.getRows(); row++)
		{
			if(sheet.getCell(0, row).getContents().equals(symbol))
			{
				startOfContent = row;
				break;
			}
		}
		if(startOfContent == -1)
			throw new Exception("Start of content could not be found!");
		
		//Zeile finden wo der Inhalt aufhoert
		int endOfContent = -1;
		for(int row = startOfContent; row < sheet.getRows(); row++)
		{
			if(!sheet.getCell(0, row).getContents().equals(symbol))
			{
				endOfContent = row;
				break;
			}
		}
		if(endOfContent == -1)
			endOfContent = sheet.getRows();
		
		
		return new XLSMap(sheet, symbol, date, startOfContent, endOfContent, pi0, qi0, qit, ffi, ci, kt, indexIsin, isin);
	}

	private void saveIndexRecord(XLSMap xlsMap, Mutator<String> mutator)
	{
		List<HColumn<String, Double>> list = new LinkedList<HColumn<String, Double>>();
		//ISIN
		String isin = xlsMap.sheet.getCell(xlsMap.indexIsinColumn, xlsMap.startLine).getContents();
		
		//Verkettungsfaktor
		double k = ((NumberCell)xlsMap.sheet.getCell(xlsMap.ktColumn, xlsMap.startLine)).getValue();
		HColumn<String, Double> c = HFactory.createColumn("Kt", k,
				StringSerializer.get(), DoubleSerializer.get());
		list.add(c);
		
		//Basis
		c = HFactory.createColumn("base", (double)BASE,
				StringSerializer.get(), DoubleSerializer.get());
		list.add(c);
		
		
		//Nun wird die Column unter der Zeit Zeile gespeichert
		HSuperColumn<String, String, Double> sc = HFactory.createSuperColumn(isin, list, StringSerializer.get(),
				StringSerializer.get(), DoubleSerializer.get());
		mutator.insert(xlsMap.date, xlsMap.symbol, sc);
	}	
}
